package org.zjt.flink.stream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 *
 *      没秒保存一次checkPoint内容，用于容错处理。
 *
 * @author juntao.zhang
 * Date: 2018-10-12 下午3:36
 * @see
 */
public class StateBackendTestBase {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                List<String> arrays = Arrays.asList("zhang", "wang", "zhang", "li");
                Random random = new Random();
                for (int j = 0; j < 100; j++) {
                    TimeUnit.SECONDS.sleep(1);
                    int i = random.nextInt(arrays.size());
                    ctx.collect(arrays.get(i));
                }
            }

            @Override
            public void cancel() {}
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value,1);
            }
        }).keyBy(0).timeWindow(Time.seconds(30))
                .apply(new WindowFunction() {
                    @Override
                    public void apply(Object o, Window window, Iterable input, Collector out) throws Exception {
                        System.out.println(window);
                        Iterator iterator = input.iterator();
                        while (iterator.hasNext()){
                            Object next = iterator.next();
                            out.collect(next);
                        }
                    }
                }).print();

        env.execute();
    }


    public static void checkPoint() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // start a checkpoint every 1000 ms  每一秒保存一次checkPoint
        env.enableCheckpointing(1000);

        // advanced options:

        // set mode to exactly-once (this is the default)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // make sure 500 ms of progress happen between checkpoints
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

        // checkpoints have to complete within one minute, or are discarded
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        // allow only one checkpoint to be in progress at the same time
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        // enable externalized checkpoints which are retained after job cancellation
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);


        env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                List<String> arrays = Arrays.asList("zhang", "wang", "zhang", "li");
                Random random = new Random();
                for (int j = 0; j < 100; j++) {
                    TimeUnit.SECONDS.sleep(1);
                    int i = random.nextInt(arrays.size());
                    ctx.collect(arrays.get(i));
                }
            }

            @Override
            public void cancel() {}
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value,1);
            }
        }).keyBy(0).timeWindow(Time.seconds(30)).sum(1).print();

        env.execute();
    }
}
